#include "kafkaProducer.h"

//生产者构造函数，给定指定的broker名称、partition数量以及topic名称
KafkaProducer::KafkaProducer(const string brokers, const string topics, int partition_size)
    :m_strBroker(brokers),m_strTopics(topics),m_partitionSize(partition_size),m_isRun(true)
{

}

KafkaProducer::~KafkaProducer(){
    stop();
}

bool KafkaProducer::init(){
    using RdKafka::Conf;

    string errstr="";

    //全局配置
    Conf* global_conf=Conf::create(Conf::ConfType::CONF_GLOBAL);

    //topic配置
    Conf* topic_conf=Conf::create(Conf::ConfType::CONF_TOPIC);

    //设置brokerlist
    if(global_conf->set("metadata.broker.list",m_strBroker,errstr)!=Conf::CONF_OK){
        std::cerr << "RdKafka conf set brokerlist failed :" << errstr << endl;
    }

    //设置回调信息
    global_conf->set("dr_cb",&m_producerDeliveryReportCallBack,errstr);
    global_conf->set("event_cb",&m_producerEventCallBack,errstr);

    //使用全局配置对象创建一个生产者
    m_ProducerPtr=RdKafka::Producer::create(global_conf,errstr);
    if(!m_ProducerPtr){
        std::cerr<<"create producer error: "<<errstr<<endl;
        return false;
    }
    
    cout<<"create producer: "<<m_ProducerPtr->name()<<" ok!"<<endl;

    //初始化topic
    m_TopicPtr=RdKafka::Topic::create(m_ProducerPtr,m_strTopics,topic_conf,errstr);
    if(!m_TopicPtr){
        std::cerr<<"create topic error: "<<errstr<<endl;
        return false;
    }
    cout<<"create topic: "<<m_TopicPtr->name()<<" ok!"<<endl;

    delete global_conf;
    delete topic_conf;

    return true;
}

//向指定的topic发送消息
void KafkaProducer::send(const string topic, const string &msg){
    using RdKafka::Conf;
    string errstr;
    //topic配置
    Conf* topic_conf=Conf::create(Conf::ConfType::CONF_TOPIC);
    //初始化topic
    m_TopicPtr=RdKafka::Topic::create(m_ProducerPtr,topic,topic_conf,errstr);
    if(!m_TopicPtr){
        std::cerr<<"(send): create topic error: "<<errstr<<endl;
        return;
    }

    this->send(msg);

    delete topic_conf;
}

void KafkaProducer::send(const string& message){
    if(m_isRun==false)
        return;
    
    //生产者生产，发送消息
    RdKafka::ErrorCode res=m_ProducerPtr->produce(m_TopicPtr,m_partitionSize,
                                            RdKafka::Producer::RK_MSG_COPY,
                                            const_cast<char*>(message.c_str()),message.size(),nullptr,nullptr);
    //有错误
    if(res!=RdKafka::ERR_NO_ERROR)
        std::cerr << "Produce failed: " << RdKafka::err2str(res) << std::endl;
    else
        std::cerr << "Produced message (" << message.size() << " bytes)" << std::endl;

    m_ProducerPtr->poll(0);

    while(m_isRun&&m_ProducerPtr->outq_len()>0){
        std::cerr<<"waiting for: "<<m_ProducerPtr->outq_len()<<endl;
        m_ProducerPtr->poll(100);
    }
}

void KafkaProducer::stop(){
    if(m_TopicPtr!=nullptr){
        delete m_TopicPtr;
        m_TopicPtr=nullptr;
    }
    
    if(m_ProducerPtr!=nullptr){
        delete m_ProducerPtr;
        m_ProducerPtr=nullptr;
    }
}

void KafkaProducer::createTopic(string topic){
    //todo 写这个创建话题的代码，然后改到chat项目里
    using RdKafka::Conf;
    string errstr;
    //topic配置
    Conf* topic_conf=Conf::create(Conf::ConfType::CONF_TOPIC);
    //初始化topic
    m_TopicPtr=RdKafka::Topic::create(m_ProducerPtr,topic,topic_conf,errstr);
    if(!m_TopicPtr){
        std::cerr<<"(createTopic): create topic error: "<<errstr<<endl;
        return;
    }
    cout<<"create topic: "<<topic<<" ok!"<<endl;

    delete topic_conf; 
    
}